/*
 * thread_pool.c
 *
 *  Created on: 2015-5-15
 *      Author: likui
 */


#include "thread_pool.h"
#include "http_conn.h"
#include "error.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <pthread.h>

extern int epollfd;

/**
 * @brief
 * @param thread_num
 * @param max_req
 * @return
 */
thread_pool_t* thread_pool_create(int thread_num, int max_req)
{
#ifndef NDEBUG
	error_msg("thread_pool_create");
#endif

	thread_pool_t *thread_pool;

	thread_pool = (thread_pool_t *)malloc(sizeof(thread_pool_t));
	if (thread_pool == NULL) {
		error_msg("malloc error");
		return NULL;
	}

	// 初始化线程池
	if (thread_pool_init(thread_pool, thread_num, max_req) == false) {
		error_msg("thread_pool_init error");
		free(thread_pool);
		return NULL;
	}

	return thread_pool;
}
/**
 * @brief 初始化线程池结构
 * @param thread_pool 线程池结构thread_pool_t 指针
 * @param thread_num 线程数
 * @param max_req 最大请求数
 * @return 若成功则返回true,若失败则返回false
 */
bool thread_pool_init(thread_pool_t *thread_pool, int thread_num, int max_req)
{
	int i;

	// 检查参数
	if ( (thread_num <= 0) || (max_req <= 0) ) {
		error_msg("thread_num = %d, max_req = %d", thread_num, max_req);
		goto done1;					// 不需要释放动态分配空间
	}

	// 动态分配线程ID空间，需要释放内存
	thread_pool->thread_pids = (pthread_t *)malloc(thread_num * sizeof(pthread_t));
	if (thread_pool->thread_pids == NULL) {
		error_ret("malloc error");
		goto done1;					// 不需要释放动态分配空间
	}
	thread_pool->thread_number = thread_num;		// 线程数

	// 请求队列，队列为动态分配的空间
	thread_pool->max_request = max_req;											// 最大请求数
	if (queue_init( &(thread_pool->request_queue), max_req) == false) {		// 初始化请求队列
		error_msg("queue_init error");
		goto done2;				// 需要释放线程ID空间
	}

	if ( (errno = pthread_mutex_init( &(thread_pool->queue_mutex), NULL)) != 0) {	// 保护请求队列的互斥锁
		error_ret("pthread_mutex_init error");
		goto done3;				// 需要释放线程ID空间、请求队列空间
	}

	// 初始化信号量，是否有请求需要处理
	if ( sem_init(&(thread_pool->sem_request), 0, 0) != 0) {
		error_ret("sem_init error");
		goto done3;				// 需要释放线程ID空间、请求队列空间
	}

	thread_pool->stop = false;						// 是否停止

	// 创建线程池
	for (i = 0; i < thread_num; i++) {
		error_msg("create the %dth thread", i);

		// 创建线程
		if ( (errno = pthread_create(&thread_pool->thread_pids[i], NULL, thread_pool_worker, thread_pool)) != 0) {
			error_ret("pthread_create error");
			goto done2;				// 需要释放线程ID空间
		}

		// 线程分离
		if ((errno = pthread_detach(thread_pool->thread_pids[i])) != 0) {
			error_ret("pthread_detach error");
			goto done2;				// 需要释放线程ID空间
		}
	}

	return true;

done3:
	queue_destroy( &(thread_pool->request_queue));	// 释放队列空间

done2:
	free(thread_pool->thread_pids);					// 释放线程ID空间
	thread_pool->thread_pids = NULL;

done1:
	return false;
}

/**
 * @brief 销毁线程池结构
 * @param thread_pool 线程池结构thread_pool_t指针
 */
void thread_pool_destroy(thread_pool_t **thread_pool)
{
	int i;
	thread_pool_t *tp = *thread_pool;

	tp->stop = true;

	sleep(1);

	tp->max_request = 0;								// 请求队列中最大请求数
	queue_destroy( &(tp->request_queue));			// 释放队列空间
	pthread_mutex_destroy( &(tp->queue_mutex));	// 销毁互斥量

	sem_destroy(&tp->sem_request);					// 销毁信号量

	for (i = 0; i < tp->thread_number; i++) {
		pthread_cancel(tp->thread_pids[i]);			// 取消其他线程
	}

	sleep(1);

	free(tp->thread_pids);							// 释放线程ID空间
	tp->thread_pids = NULL;
	tp->thread_number = 0;							// 线程数



	free(tp);
	*thread_pool = NULL;
}

/**
 * @brief 添加HTTP请求http_conn到队列到线程池变量thread_pool
 * @param thread_pool 线程池变量thread_pool_t指针
 * @param http_conn HTTP遍历http_conn_t 指针
 * @return 若添加成功则返回true,若添加失败则返回false
 */
bool thread_pool_append(thread_pool_t *thread_pool ,http_conn_t *http_conn)
{
#ifndef NDEBUG
	error_msg("thread_pool_append start: %lx", pthread_self());
#endif
	// 队列为所有线程共享，需要加锁
	pthread_mutex_lock( &(thread_pool->queue_mutex));

	if (queue_is_full( &(thread_pool->request_queue))) {
		// 请求队列已满，添加失败
		pthread_mutex_unlock( &(thread_pool->queue_mutex));
		return false;
	}

	// 添加元素到队列
	queue_add( &(thread_pool->request_queue), http_conn);

	pthread_mutex_unlock( &(thread_pool->queue_mutex));

	// post信号量
	sem_post( &(thread_pool->sem_request));
#ifndef NODEBUG
	printf("post: %lx\n", pthread_self());
#endif

	return true;
}

/**
 * @brief 工作线程启动时入口函数
 * @param arg 线程池结构thread_pool_t指针
 * @return 返回线程池thread_pool 变量指针
 */
void *thread_pool_worker(void *arg)
{
	thread_pool_t *thread_pool = (thread_pool_t *)arg;

	thread_pool_run(thread_pool);

	return thread_pool;
}

/**
 * @brief 工作线程等待或处理请求
 * @param thread_pool 线程池结构thread_pool_t指针
 */
void thread_pool_run(thread_pool_t *thread_pool)
{
#ifndef NDEBUG
	error_msg("thread_pool_run start: %lx", pthread_self());
#endif
	while (thread_pool->stop == false) {
		sem_wait( &(thread_pool->sem_request));		// 等待请求到来

		pthread_mutex_lock( &(thread_pool->queue_mutex));

#ifndef NDEBUG
	printf("wakeup: pid = %lx\n", pthread_self());
#endif

		if (queue_is_empty( &(thread_pool->request_queue))) {
			pthread_mutex_unlock( &(thread_pool->queue_mutex));
			continue;
		}

		http_conn_t *http_conn = queue_del( &(thread_pool->request_queue));

		pthread_mutex_unlock( &(thread_pool->queue_mutex));

		http_conn_process(http_conn, epollfd);
	}
}
